package com.itcast.spark

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * DESC:实现类类似于百度热词统计，这里统计的是每个5s处理近10s的数据--window
 * 1-准备环境
 * 2-读取数据
 * 3-窗口的统计
 * 4-结果输出print
 * 5-ssc.start
 * 6-ssc.awaitTermination
 * 7-ssc.stop(true,true)
 */
object SparkStreamingTCPTopK {
  def main(args: Array[String]): Unit = {
    //1-准备环境
    val conf: SparkConf = new SparkConf().setAppName("SparkStreamingTCPTopK").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //这里就是指定配置项将数据按照5秒为周期进行处理
    val ssc = new StreamingContext(sc, Seconds(5))
    //2-读取数据
    val receiveData: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 9999)
    //3-窗口的统计
    val result: DStream[(String, Int)] = receiveData.flatMap(_.split(" ")).map(x => (x, 1)).reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(10), Seconds(5))
    //4-结果输出print
    result.print()
    //5-ssc.start
    ssc.start()
    //6-ssc.awaitTermination
    ssc.awaitTermination()
    //7-ssc.stop(true,true)
    ssc.stop(true, true)
  }
}
